package com.autonavi.collapse.task;

import com.autonavi.collapse.concurrent.AsyncScheduledThreadPool;
import com.autonavi.collapse.task.context.TaskComponentContainer;
import com.autonavi.collapse.task.context.utils.NetUtils;
import com.autonavi.collapse.task.dispatch.DispatchContext;
import com.autonavi.collapse.task.dispatch.MinimumTaskDispatcher;
import com.autonavi.collapse.task.dispatch.TaskDispatcher;
import com.autonavi.collapse.task.dispatch.TaskNode;
import com.autonavi.collapse.task.leader.LeaderSelector;
import com.autonavi.collapse.task.model.TaskConfig;
import com.autonavi.collapse.task.model.TaskRelation;
import com.autonavi.collapse.task.model.TaskTemplate;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class TaskContainer {

    private static final Logger LOGGER = LoggerFactory.getLogger(TaskContainer.class);
    private List<TaskListener> taskListeners = new ArrayList<>();
    private TaskService taskService;
    private LeaderSelector leaderSelector;
    private TaskDispatcher taskDispatcher;

    private ScheduledExecutorService pullScheduledExecutorService;
    private ScheduledExecutorService createScheduledExecutorService;
    private ScheduledExecutorService dispatchScheduledExecutorService;

    private String myName;
    private int taskFetchSize = 8;
    private int taskDispatchSize = 32;
    private int templateFetchSize = 8;
    private int interval = 5000;
    private static final int RELATION_LIMIT_SIZE = 100;
    private Set<String> disPatchedTaskIds = new HashSet<>();
    private List<String> typeList;

    public void init() throws Exception {
        if (StringUtils.isBlank(myName)) {
            myName = NetUtils.getLocalIp();
        }
        pullScheduledExecutorService = new AsyncScheduledThreadPool(1);
        createScheduledExecutorService = new AsyncScheduledThreadPool(1);
        dispatchScheduledExecutorService = new AsyncScheduledThreadPool(1);
    }

    public void start(String group, String dataCenterName) throws Exception {
        Object dataCenter = TaskComponentContainer.getComponent(dataCenterName);
        if (dataCenter == null) {
            throw new IllegalArgumentException("DataCenter not found");
        }
        if (taskService == null) {
            taskService = TaskComponentContainer.getComponent(TaskService.class);
            if (taskService == null) {
                throw new IllegalArgumentException("TaskService not found");
            }
            taskService.init(group, dataCenter);
        }
        if (leaderSelector == null) {
            leaderSelector = TaskComponentContainer.getComponent(LeaderSelector.class);
            if (leaderSelector == null) {
                throw new IllegalArgumentException("LeaderSelector not found");
            }
            leaderSelector.init(group, dataCenter);
        }
        if (taskDispatcher == null) {
            taskDispatcher = new MinimumTaskDispatcher();
        }
        TaskRuntimeManager taskRuntimeManager = new TaskRuntimeManager(myName);
        taskRuntimeManager.setTaskService(taskService);
        taskListeners.add(taskRuntimeManager);

        for (TaskListener taskListener : taskListeners) {
            taskListener.init();
        }

        leaderSelector.start();

        pullScheduledExecutorService.scheduleAtFixedRate(new PullTask(), interval, interval, TimeUnit.MILLISECONDS);
        LOGGER.info("Task fetcher start run");
        if (leaderSelector != null) {
            createScheduledExecutorService.scheduleAtFixedRate(new CreateTask(), interval, interval, TimeUnit.MILLISECONDS);
            LOGGER.info("Task generator start run");
            dispatchScheduledExecutorService.scheduleAtFixedRate(new DispatchTask(), interval, interval, TimeUnit.MILLISECONDS);
            LOGGER.info("Task dispatcher start run");
        }
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            pullScheduledExecutorService.shutdown();
            createScheduledExecutorService.shutdown();
            dispatchScheduledExecutorService.shutdown();
            leaderSelector.stop();
            for (TaskListener taskListener : taskListeners) {
                taskListener.destroy();
            }
            LOGGER.info("collapse task container destroy");
        }, "collapse-task-container-destroy"));
        LOGGER.info("collapse-task-container-start");
    }

    public void addListener(TaskListener taskListener) {
        if (taskListener != null) {
            taskListeners.add(taskListener);
        }
    }

    public LeaderSelector getLeaderSelector() {
        return leaderSelector;
    }

    public void setLeaderSelector(LeaderSelector leaderSelector) {
        this.leaderSelector = leaderSelector;
    }

    public TaskService getTaskService() {
        return taskService;
    }

    public void setTaskService(TaskService taskService) {
        this.taskService = taskService;
    }

    public int getTaskFetchSize() {
        return taskFetchSize;
    }

    public void setTaskFetchSize(int taskFetchSize) {
        this.taskFetchSize = taskFetchSize;
    }

    public int getTemplateFetchSize() {
        return templateFetchSize;
    }

    public void setTemplateFetchSize(int templateFetchSize) {
        this.templateFetchSize = templateFetchSize;
    }

    public int getInterval() {
        return interval;
    }

    public void setInterval(int interval) {
        this.interval = interval;
    }

    public Set<String> getDisPatchedTaskIds() {
        return disPatchedTaskIds;
    }

    public void setDisPatchedTaskIds(Set<String> disPatchedTaskIds) {
        this.disPatchedTaskIds = disPatchedTaskIds;
    }

    private class PullTask implements Runnable {

        @Override
        public void run() {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("{} start pull tasks", myName);
            }
            List<TaskConfig> tasks = null;
            try {
                tasks = taskService.findMyTask(myName, taskFetchSize);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
            if (tasks != null && !tasks.isEmpty()) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("{} pull {} tasks", myName, tasks.size());
                }
                Set<String> lastTaskIds = new HashSet<>(disPatchedTaskIds);
                Set<String> currentTaskIds = new HashSet<>();
                for (TaskConfig task : tasks) {
                    lastTaskIds.remove(task.getId());
                    currentTaskIds.add(task.getId());
                    for (TaskListener taskListener : taskListeners) {
                        taskListener.onDispatch(task);
                    }
                }
                for (String laseTaskId : lastTaskIds) {
                    for (TaskListener taskListener : taskListeners) {
                        taskListener.onCancel(laseTaskId);
                    }
                }
                disPatchedTaskIds = currentTaskIds;
            }
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("{} finish pull tasks", myName);
            }
        }
    }

    private class CreateTask implements Runnable {

        @Override
        public void run() {
            if (!leaderSelector.isLeader()) {
                return;
            }
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("{} start generate tasks ", myName);
            }
            List<TaskTemplate> templates = null;
            try {
                templates = taskService.findTemplates(templateFetchSize);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
            if (templates != null) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("{} find {} templates", myName, templates.size());
                }
                for (TaskTemplate template : templates) {
                    try {
                        List<TaskConfig> taskConfigs = new ArrayList<>();
                        for (TaskListener taskListener : taskListeners) {
                            List<TaskConfig> temp = taskListener.onCreate(template);
                            if (temp != null) {
                                taskConfigs.addAll(temp);
                            }
                        }
                        taskService.addTask(taskConfigs);
                        Calendar now = Calendar.getInstance();
                        now.add(Calendar.SECOND, template.getInterval());
                        template.setNextTime(now.getTime());
                        taskService.saveTemplateNextTime(template);
                        if (!taskConfigs.isEmpty()) {
                            LOGGER.info("Template {} create {} task", template.getName(), taskConfigs.size());
                        }
                    } catch (Exception e) {
                        LOGGER.error(e.getMessage(), e);
                    }
                }
            }
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("{} finish generate tasks", myName);
            }
        }
    }

    private class DispatchTask implements Runnable {

        @Override
        public void run() {
            if (!leaderSelector.isLeader()) {
                return;
            }
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("{} start dispatch tasks", myName);
            }
            try {
                List<String> types = taskService.findTaskTypeList();
                if (types != null && types.size() > 0) {
                    for (String taskType : types) {
                        List<TaskNode> availableNodes = taskService.findAvailableNodes();
                        DispatchContext context = new DispatchContext();
                        context.setAvailableNodes(availableNodes);
                        int availableNodeSize = availableNodes.size();
                        if (availableNodeSize == 0) {
                            LOGGER.warn("No available task node");
                            return;
                        }
                        List<TaskRelation> taskRelations = taskService.FindtaskRelations(RELATION_LIMIT_SIZE);
                        if (taskRelations != null && !taskRelations.isEmpty()) {
                            context.setRelations(taskRelations);
                        }
                        List<TaskConfig> deadTasks = new ArrayList<>();
                        List<TaskNode> deadNodes = taskService.findDeadNodes();
                        for (TaskNode deadNode : deadNodes) {
                            deadTasks.addAll(taskService.findMyTask(deadNode.getName(), taskType, deadNode.getFreeSize()));
                        }
                        context.setDeadTasks(deadTasks);
                        context.setTasks(taskService.findTasks(taskDispatchSize, taskType));

                        List<TaskConfig> changedConfigs = taskDispatcher.dispatch(context);
                        if (changedConfigs != null && !changedConfigs.isEmpty()) {
                            LOGGER.info("Dispatch tasks:[{}]", changedConfigs);
                        }
                        if (changedConfigs != null) {
                            for (TaskConfig changedConfig : changedConfigs) {
                                taskService.dispatch(changedConfig);
                            }
                        }
                    }
                }
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("{} finish dispatch tasks", myName);
            }
        }
    }
}
